package cn.whuc.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Demo_Accumulator {
  def main(args: Array[String]): Unit = {
    // 1 创建sparkContext
    val sc: SparkContext = new SparkContext(
      new SparkConf()
        .setMaster("local[*]")
        .setAppName(" ")
    )

    // 2 编写代码
    //使用累加器 完成词频统计
    val rdd: RDD[String] = sc.makeRDD(List("hello", "spark", "hadoop", "spark"))

    val acc = new MyAccumutor()

    // 注册
    sc.register(acc)

    // 使用
    rdd.foreach(word=>{
      acc.add(word)
    })

    println(acc.value)

    // 3 关闭上下文对象
    sc.stop()
  }
}

// 累加器输出需要可变集合中的map对象
// 在当前案例中 设计累加器的输出是map，方便处理
// 实际上，对于输出结构可以自行设计，回顾aggregateByKey算子中
// 提到的零值 虽然累加器和零值没有直接的关系 但变换结构输出的思路 如出一辙
class MyAccumutor extends AccumulatorV2[String,mutable.Map[String,Long]] {
  // 声明可变集合的map对象
  var map = mutable.Map[String,Long]()

  // 判断集合是否位空
  override def isZero: Boolean = map.isEmpty

  // 创建当前类对象
  override def copy(): AccumulatorV2[String, mutable.Map[String, Long]] = new MyAccumutor

  // 清空 重置
  override def reset(): Unit = map.clear()

  // 每一个excutor中的累加器逻辑操作
  override def add(v: String): Unit = {
    // 判断map中有没有当前的V 有则count+1 无则计1
    val count: Long = map.getOrElse(v, 0L) + 1
    // 更新v的计数
    map.update(v,count)
  }

  // 不用excutor中的 acc的聚合逻辑
  override def merge(other: AccumulatorV2[String, mutable.Map[String, Long]]): Unit = {
    val map1: mutable.Map[String, Long] = this.map
    val map2: mutable.Map[String, Long] = other.value
    // 当前的map中 有无mao2的元素 有则聚合 无就计0
    map2.foreach{
      case (word,count) =>{
        // map1和map2中同一个单词的count的累加
        val newcount: Long = map1.getOrElse(word, 0L)+count
        // 更新map1
        map1.update(word,newcount)
      }
    }
  }

  override def value: mutable.Map[String, Long] = map
}